package com.xzx.flink.streamapi.transform;

import com.xzx.flink.bean.ClickEvent;
import com.xzx.flink.streamapi.source.ClickSource;
import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/4/17 21:45
 */
public class Transform_11_PartitionCustom {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStream<ClickEvent> source = env.addSource(new ClickSource())
                .partitionCustom(new Partitioner<String>() {
                    @Override
                    public int partition(String key, int numPartitions) {
                        return key.hashCode() % numPartitions;
                    }
                }, new KeySelector<ClickEvent, String>() {
                    @Override
                    public String getKey(ClickEvent value) throws Exception {
                        return value.getUser();
                    }
                });
        source.print();

        env.execute(Transform_11_PartitionCustom.class.getSimpleName());
    }
}
